package com.zimin.rocketmq.utils;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * rocket mq 工具类
 *
 * @author 恒利
 * @since 2022/6/29 10:21 PM
 */
@Slf4j
public class RocketMQUtil {

    /**
     * <p>同步发送消息</p>
     *
     * @param producer 生产者
     * @param msg      发送的消息内容
     * @since 2022-06-29 22:23:10
     */
    public static void syncSendMsg(DefaultMQProducer producer, Message msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        var result = producer.send(msg);
        log.info(JSON.toJSONString(result));
    }

    /**
     * <p>异步发送消息</p>
     *
     * @param producer 生产者
     * @param msg      发送的消息内容
     * @since 2022-06-30 10:43:30
     */
    public static void asyncSendMsg(DefaultMQProducer producer, Message msg) throws RemotingException, InterruptedException, MQClientException {
        int messageCount = 2;
        var countDownLatch = new CountDownLatch2(messageCount);
        for (int i = 0; i < messageCount; i++) {
            producer.send(msg, new SendCallback() {

                @Override
                public void onSuccess(SendResult sendResult) {
                    countDownLatch.countDown();
                    log.info(sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable throwable) {
                    countDownLatch.countDown();
                    log.info("发送消息时发生了异常!" + throwable);
                    throwable.printStackTrace();
                }
            });
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
    }

}
